package com.flink;

import cn.hutool.core.date.DatePattern;
import com.alibaba.fastjson2.JSON;
import com.kafka.KafkaProducerService;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Random;

@Component
public class Test1 {

    @Resource
    private KafkaProducerService kafkaProducerService;

    private Random random = new Random();

    @PostConstruct
    public void init() {
        //this.createMsg2Kafka();
        this.createClickEvents();
    }

    public void createClickEvents(){
        while (true){
            int i = random.nextInt(10);
            int p = i + 1;
            ClickEvents events = ClickEvents.builder().userId("user_"+p).eventTime(LocalDateTime.now().format(DatePattern.NORM_DATETIME_MS_FORMATTER)).pageUrl("/test_page_"+p).build();
            kafkaProducerService.sendMessage("user-clicks", JSON.toJSONString(events));
            try {
                Thread.sleep(1000L * p);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

    }

    public void createMsg2Kafka() {

        while (true) {
            int nextInt = Math.abs(random.nextInt());
            int sub = random.nextInt(10);
            PointData p = PointData.builder().id(System.currentTimeMillis()).deviceCode("设备"+sub).pointType("点位"+sub).value(nextInt).collectTime(LocalDateTime.now()).build();
            kafkaProducerService.sendMessage("t1", JSON.toJSONString(p));
            try {
                Thread.sleep(300L);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}
